package org.example.reactor;

import org.example.utils.LogUtils;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 多线程处理模式
 * 一个线程负责处理Socket连接
 * 多个线程负责处理Socket传输
 */
public class MultiThreadServerSocket {

    private static Selector[] workSelectors;

    private static Selector bossSelector;

    private static AtomicInteger indexs = new AtomicInteger(0);

    public static void main(String[] args) throws IOException{

        bossSelector = Selector.open();
        LogUtils.info("bossSelector=====>" + bossSelector);
        ServerSocketChannel serverChannel = ServerSocketChannel.open();
        serverChannel.configureBlocking(false);
        serverChannel.bind(new InetSocketAddress("localhost", 8081));
        SelectionKey skey = serverChannel.register(bossSelector, 0);
        skey.attach(new MultiAcceptHandler(skey));
        skey.interestOps(SelectionKey.OP_ACCEPT);

        workSelectors = new Selector[2];
        workSelectors[0] = Selector.open();
        LogUtils.info("workSelectors[0]=====>" + workSelectors[0]);
        workSelectors[1] = Selector.open();
        LogUtils.info("workSelectors[1]=====>" + workSelectors[1]);
        Reactor bossReactor = new Reactor(bossSelector);
        Reactor workReactor1 = new Reactor(workSelectors[0]);
        Reactor workReactor2 = new Reactor(workSelectors[1]);
        new Thread(bossReactor).start();
        // 线程1负责Socket传输
        new Thread(workReactor1).start();
        // 线程2负责Socket传输
        new Thread(workReactor2).start();
    }


    /**
     * 反应器线程
     */
    public static class Reactor implements Runnable {

        private Selector selector;

        public Reactor(Selector selector) {
            this.selector = selector;
        }

        @Override
        public void run() {
            try {
                while (!Thread.interrupted()) {
//                    LogUtils.info(selector.toString());
                    selector.select(1000);
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    if (null == selectionKeys || selectionKeys.size() == 0) {
                        continue;
                    }
                    Iterator<SelectionKey> iterator = selectionKeys.iterator();
                    while (iterator.hasNext()) {
                        SelectionKey selectionKey = iterator.next();
                        dispatch(selectionKey);
//                        iterator.remove();
                    }
                    selectionKeys.clear();
                }
            } catch (IOException e) {
                LogUtils.info(e.getMessage());
            }
        }

        public void dispatch(SelectionKey skey) {
            Runnable handler = (Runnable) skey.attachment();
            if (handler != null) {
                handler.run();
            }
        }
    }

    public static class MultiAcceptHandler implements Runnable {

        private SelectionKey skey;

        public MultiAcceptHandler(SelectionKey skey) {
            this.skey = skey;
        }

        @Override
        public void run() {
            try {
                ServerSocketChannel serverChannel = (ServerSocketChannel) skey.channel();
                Selector selector = workSelectors[indexs.get()];
                SocketChannel socketChannel = serverChannel.accept();
                if (socketChannel != null) {
                    new MultiThreadHandler(selector, socketChannel);
                }

                // 进行轮询
                if (indexs.incrementAndGet() == workSelectors.length) {
                    indexs.set(0);
                }
            } catch (IOException e) {
                LogUtils.info(e.getMessage());
            }
        }
    }
}
